package com.cv.infra.service.dynamicThreadPool;

import cn.hutool.core.util.ObjUtil;
import com.cv.framework.common.exception.ServerException;
import com.cv.framework.common.exception.ServiceException;
import com.cv.framework.common.exception.constants.GlobalErrorCodeConstants;
import com.cv.infra.api.file.dto.DynamicThreadDTO;
import com.cv.infra.controller.system.dynamicThreadPool.vo.DynamicThreadPoolReqVO;
import com.cv.infra.controller.system.dynamicThreadPool.vo.DynamicThreadPoolRespVO;
import com.cv.infra.framework.dynamicThreadPool.config.DynamicThreadPoolConfiguration;
import com.cv.infra.framework.dynamicThreadPool.core.DynamicQueue;
import com.cv.infra.service.file.FileService;
import com.cv.platform.api.video.CvVideoApi;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.minio.BucketExistsArgs;
import io.minio.MakeBucketArgs;
import io.minio.MinioClient;
import io.minio.PutObjectArgs;
import io.minio.errors.MinioException;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.*;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import static com.cv.infra.dal.redis.RedisKeyConstants.DYNAMIC_THREAD_POOL_KEY;

/**
 * 动态线程池业务层接口实现类
 *
 * @author Charles_XDXD
 */
@Slf4j
@Service
public class DynamicThreadPoolServiceImpl implements DynamicThreadPoolService {

    @Resource(name = "dynamicThreadPool")
    private ThreadPoolExecutor dynamicThreadPool;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    @Lazy
    private FileService fileService;

    @Resource
    private MinioClient minioClient;

    @Value("${minio.endpoint}")
    private String endpoint;

    @Value("${minio.bucket-name}")
    private String minioBucket;

    @Value("${os.win.localStorage.transcodePath}")
    String winTranscodePath;

    @Value("${os.linux.localStorage.transcodePath}")
    private String linuxTranscodePath;

    @Value("${os.mac.localStorage.transcodePath}")
    private String macTranscodePath;

    @DubboReference
    private CvVideoApi cvVideoApi;

    /**
     * 获取动态线程池信息
     *
     * @return 包含线程池和系统信息的响应对象
     * @throws ServiceException 如果线程池已经关闭或正在终止
     */
    @Override
    public DynamicThreadPoolRespVO getDynamicThreadPoolInfo() {
        if (dynamicThreadPool.isShutdown() || dynamicThreadPool.isTerminated() || dynamicThreadPool.isTerminating()) {
            throw new ServerException(GlobalErrorCodeConstants.INTERNAL_SERVER_ERROR);
        }

        return new DynamicThreadPoolRespVO(dynamicThreadPool);
    }

    /**
     * 修改动态线程池配置
     *
     * @param dynamicThreadPoolReqVO 配置请求对象
     */
    @Override
    public void modifyDynamicThreadPool(DynamicThreadPoolReqVO dynamicThreadPoolReqVO) {
        RLock lock = redissonClient.getLock(DYNAMIC_THREAD_POOL_KEY);

        try {
            // 加锁，预防在修改配置时，有任务进入
            lock.lock();

            // 获取当前队列中的任务数量
            int currentQueueSize = dynamicThreadPool.getQueue().size();

            // 获取新的队列最大容量
            int newQueueCapacity = dynamicThreadPoolReqVO.getQueueCapacity();

            DynamicQueue<Runnable> queue = (DynamicQueue<Runnable>) dynamicThreadPool.getQueue();

            // 保证当前队列是空的
            if (queue.getCapacity() != newQueueCapacity && currentQueueSize > 0) {
                throw new ServiceException("当前队列中存在未处理的任务，无法修改队列容量。当前任务数: " + currentQueueSize);
            }

            // 检查当前队列中的任务数量是否超过新配置的队列容量
            if (currentQueueSize > newQueueCapacity) {
                throw new ServiceException("当前队列中的任务数量超过新配置的队列容量。当前任务数: " + currentQueueSize + ", 新队列容量: " + newQueueCapacity);
            }

            // 修改线程池配置
            if (queue.getCapacity() != newQueueCapacity) {
                queue.setCapacity(newQueueCapacity);
            }

            dynamicThreadPool.setCorePoolSize(dynamicThreadPoolReqVO.getCorePoolSize());
            dynamicThreadPool.setMaximumPoolSize(dynamicThreadPoolReqVO.getMaximumPoolSize());
            dynamicThreadPool.setKeepAliveTime(dynamicThreadPoolReqVO.getKeepAliveTime(), TimeUnit.SECONDS);

            log.info("[线程池监控] 动态线程池配置已更新。新的核心线程数: {}, 新的最大线程数: {}, 新的队列容量: {}, 新的保持活跃时间: {} s",
                    dynamicThreadPoolReqVO.getCorePoolSize(),
                    dynamicThreadPoolReqVO.getMaximumPoolSize(),
                    newQueueCapacity,
                    dynamicThreadPoolReqVO.getKeepAliveTime()
            );
        } finally {
            lock.unlock();
        }
    }

    /**
     * 监听当前线程池的信息和状态
     * 负载因子：90%
     */
    @Override
    public void listenDynamicThreadPool() {
        // 获取当前活动线程的数量
        int activeCount = dynamicThreadPool.getActiveCount();
        // 获取队列中的任务数量
        int queueSize = dynamicThreadPool.getQueue().size();
        // 获取线程池的最大线程数
        int maximumPoolSize = dynamicThreadPool.getMaximumPoolSize();
        // 获取线程池的核心线程数
        int corePoolSize = dynamicThreadPool.getCorePoolSize();
        // 计算总的任务数（正在运行的+等待的）
        int totalTasks = activeCount + queueSize;
        // 计算最大可接受的任务数（最大线程数 + 队列剩余容量）
        int maxTasks = maximumPoolSize + dynamicThreadPool.getQueue().remainingCapacity();

        // 计算当前线程池的负载因子
        double loadFactor = (double) totalTasks / maxTasks;

        if (loadFactor > 0.9) {
            // 尝试增加核心线程数，但不超过最大线程数
            int newCorePoolSize = Math.min(corePoolSize + 1, maximumPoolSize);
            // 尝试增加最大线程数，但不超过核心线程数的两倍
            int newMaximumPoolSize = Math.min(maximumPoolSize + 1, 2 * corePoolSize);

            dynamicThreadPool.setCorePoolSize(newCorePoolSize);
            dynamicThreadPool.setMaximumPoolSize(newMaximumPoolSize);

            log.info("[线程池监控] 线程池负载超过 90%，已增加线程池大小。新的核心线程数: {}, 新的最大线程数: {}",
                    newCorePoolSize, newMaximumPoolSize);
        }

        if (loadFactor < 0.6 && corePoolSize != DynamicThreadPoolConfiguration.CORE_POOL_SIZE) {
            dynamicThreadPool.setCorePoolSize(DynamicThreadPoolConfiguration.CORE_POOL_SIZE);

            log.info("[线程池监控] 线程池负载低于 60%，已调整回标准核心数。新的核心线程数: {}", DynamicThreadPoolConfiguration.CORE_POOL_SIZE);
        }

        // 如果当前队列占用比例超过 80%，则自动扩容
        DynamicQueue queue = (DynamicQueue) dynamicThreadPool.getQueue();
        if ((double) queueSize / queue.getCapacity() > 0.8) {
            queue.setCapacity(queueSize * 2);
            log.info("[线程池监控] 当前队列占比超 80%，已增加队列容量。新的队列容量: {}", queue.getCapacity());
        }

    }

    /**
     * 执行转码任务
     *
     * @param dynamicThreadDTO 视频转码对象
     */
    @Override
    public void executeConvertVideoTask(DynamicThreadDTO dynamicThreadDTO) {
        // 异步执行任务
        CompletableFuture<DynamicThreadDTO> future = CompletableFuture.supplyAsync(() -> {
            log.info("[DynamicThreadPool] 正在执行转码服务: ");
            transcodeVideo(dynamicThreadDTO);
            return dynamicThreadDTO;
        }, dynamicThreadPool);

        future.thenAccept(result -> {
            log.info("[DynamicThreadPool] 转码成功，开始转存到 MinIO");
            String fileName = result.getFileName();
            String randomFileName = fileName.substring(0, fileName.lastIndexOf("."));
            String minioPath = endpoint + "/" + minioBucket + "/video/" + randomFileName;

            // 遍历文件夹，上传到 MinIO
            String transcodePath = result.getTranscodePath();
            try {
                File[] files = new File(transcodePath).listFiles();
                if (ObjUtil.isNotEmpty(files)) {
                    for (File file : files) {
                        String objectName = "video/" + randomFileName + "/" + file.getName();

                        boolean exists = minioClient.bucketExists(BucketExistsArgs.builder().bucket(minioBucket).build());
                        if (!exists) {
                            minioClient.makeBucket(MakeBucketArgs.builder().bucket(minioBucket).build());
                        }

                        PutObjectArgs args = PutObjectArgs.builder()
                                .bucket(minioBucket)
                                .object(objectName)
                                .stream(new FileInputStream(file), file.length(), -1)
                                .build();
                        minioClient.putObject(args);
                    }
                }
                fileService.updateFileInfo(result.getId(), minioPath);
                log.info("[FileService] 文件成功转存到 MinIO，文件路径：{}", minioPath);

                // 更新视频转码状态为待审核
                cvVideoApi.updateTranscodingToAuditing(result.getId());
            } catch (MinioException | IOException | InvalidKeyException | NoSuchAlgorithmException e) {
                log.error("[FileService] 文件上传到 MinIO 失败", e);
                throw new RuntimeException("文件上传到 MinIO 失败", e);
            }
        }).exceptionally(exception -> {
            log.info("[DynamicThreadPool] 视频转码失败", exception);
            return null;
        });
    }

    // =================== 私有方法 ===================
    private void transcodeVideo(DynamicThreadDTO dynamicThreadDTO) {
        String transcodePath;
        String fileName = dynamicThreadDTO.getFileName();
        String filePath = dynamicThreadDTO.getFilePath();
        String transcodeFileName = fileName.substring(0, fileName.lastIndexOf("."));

        String os = System.getProperty("os.name").toLowerCase();
        if (os.contains("win")) {
            transcodePath = winTranscodePath;
        } else if (os.contains("mac")) {
            transcodePath = linuxTranscodePath;
        } else {
            transcodePath = macTranscodePath;
        }

        File transcodeFolder = new File(transcodePath);
        if (!transcodeFolder.exists()) {
            transcodeFolder.mkdirs();
        }

        String outPutFolder = transcodePath + "/" + fileName;
        if (!new File(outPutFolder).exists()) {
            new File(outPutFolder).mkdirs();
        }

        dynamicThreadDTO.setTranscodePath(outPutFolder);

        // 生成输出文件的路径，分别为 720p 和 1080p 的 M3U8 文件
        String output720p = outPutFolder + "/" + transcodeFileName + "_720p.m3u8";
        String output1080p = outPutFolder + "/" + transcodeFileName + "_1080p.m3u8";

        executeSingleTranscode(filePath, output720p, "1280x720", "5000k", "10000k");
        executeSingleTranscode(filePath, output1080p, "1920x1080", "8000k", "16000k");
    }

    /**
     * 执行单个视频转码
     */
    private void executeSingleTranscode(String inputPath, String outputPath, String resolution, String bitrate, String bufsize) {
        try {
            String[] command;

            if (hasAudioStream(inputPath)) {
                command = new String[]{"ffmpeg", "-y", "-i", inputPath,
                        "-map", "0:v", "-c:v", "libx264",
                        "-b:v", bitrate, "-maxrate", bitrate,
                        "-bufsize", bufsize, "-preset", "veryslow",
                        "-crf", "18", "-s", resolution, "-g", "60",
                        "-map", "0:a", "-c:a", "copy",
                        "-hls_time", "10", "-hls_list_size", "0",
                        "-f", "hls", outputPath
                };
            } else {
                command = new String[]{
                        "ffmpeg", "-y", "-i", inputPath,
                        "-map", "0:v", "-c:v", "libx264",
                        "-b:v", bitrate, "-maxrate", bitrate,
                        "-bufsize", bufsize, "-preset", "veryslow",
                        "-crf", "18", "-s", resolution, "-g", "60",
                        "-hls_time", "10", "-hls_list_size", "0",
                        "-f", "hls", outputPath
                };
            }

            ProcessBuilder processBuilder = new ProcessBuilder(command);
            Process process = processBuilder.start();

            // 处理输出流
            try (BufferedReader stdoutReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                 BufferedReader stderrReader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {

                CompletableFuture.runAsync(() -> {
                    stdoutReader.lines().forEach(line -> log.info("FFmpeg stdout: {}", line));
                }, dynamicThreadPool);

                CompletableFuture.runAsync(() -> {
                    stderrReader.lines().forEach(line -> log.info("FFmpeg stdout: {}", line));
                }, dynamicThreadPool);

                int exitCode = process.waitFor();
                if (exitCode == 0) {
                    log.info("转码成功: {}", outputPath);
                } else {
                    log.error("视频转码失败");
                }
            }
        } catch (Exception e) {
            log.error("视频转码发生异常: ", e);
            // 恢复中断状态
            Thread.currentThread().interrupt();
        }
    }


    /**
     * 判断是否存在音频流
     */
    private static boolean hasAudioStream(String inputPath) throws Exception {
        String[] command = {"ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", inputPath};
        ProcessBuilder pb = new ProcessBuilder(command);
        Process process = pb.start();

        // 解析 JSON 输出
        ObjectMapper mapper = new ObjectMapper();
        JsonNode rootNode = mapper.readTree(process.getInputStream());

        // 遍历所有流并检查 codec_type 是否为 audio
        if (rootNode.has("streams")) {
            for (JsonNode stream : rootNode.get("streams")) {
                if ("audio".equals(stream.get("codec_type").asText())) {
                    return true;
                }
            }
        }

        return false;
    }

}
